﻿//using System;
//using System.Collections.Generic;
//using System.Linq;
//using System.Text;

//namespace ActiveMQ.Consumer
//{
//    using Apache.NMS;
//    using Apache.NMS.ActiveMQ;
//    using Apache.NMS.ActiveMQ.Commands;
//    using System;
//    using System.Collections.Generic;
//    using System.Linq;
//    using System.Text;
//    using System.Threading;

//    namespace ActiveMQDemo
//    {
//        class Program
//        {
//            static void Main(string[] args)
//            {
//                //retriveMessage();
//                //sendData();
//                //retriveData();

//                //producer_topic();
//                retrive_topic();
//            }

//            static void producer_topic()
//            {
//                var factory = new ConnectionFactory("tcp://localhost:61616");

//                using (IConnection connection = factory.CreateConnection())
//                {
//                    using (var session = connection.CreateSession())
//                    {

//                        //var dest = session.GetDestination("first_queue", DestinationType.Queue);
//                        var producer = session.CreateProducer(new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("masterdata.material"));

//                        for (int i = 0; i < 10000; i++)
//                        {
//                            var message = producer.CreateObjectMessage(new UserInfo
//                            {
//                                ID = i + 1,
//                                Username = "张三" + i.ToString()
//                            });
//                            message.Properties.SetString("property", "property");
//                            producer.Send(message);
//                            Thread.Sleep(2000);
//                            Console.WriteLine("send " + i.ToString() + " success");
//                        }
//                    }
//                }
//            }

//            static void retrive_topic()
//            {
//                var factory = new ConnectionFactory("tcp://localhost:61616");

//                using (IConnection connection = factory.CreateConnection())
//                {
//                    connection.ClientId = "retriver-01";
//                    connection.Start();

//                    using (var session = connection.CreateSession())
//                    {
//                        session.Recover();
//                        var consumer = session.CreateDurableConsumer(new ActiveMQTopic("masterdata.material"), "listener 1", null, false);
//                        consumer.Listener += consumer_Listener;

//                        Console.ReadLine();
//                    }
//                }
//            }

//            static void sendData()
//            {
//                var factory = new ConnectionFactory("tcp://localhost:61616");

//                using (IConnection connection = factory.CreateConnection())
//                {
//                    using (var session = connection.CreateSession())
//                    {

//                        var dest = session.GetDestination("first_queue", DestinationType.Queue);
//                        var producer = session.CreateProducer(dest);
//                        for (int i = 0; i < 10000; i++)
//                        {
//                            var message = producer.CreateObjectMessage(new UserInfo
//                            {
//                                ID = i + 1,
//                                Username = "张三" + i.ToString()
//                            });
//                            message.Properties.SetString("property", "property");
//                            producer.Send(message);
//                            Thread.Sleep(500);
//                            Console.WriteLine("send " + i.ToString() + " success");
//                        }
//                    }
//                }
//            }

//            static void retriveData()
//            {
//                var factory = new ConnectionFactory("tcp://localhost:61616");

//                using (IConnection connection = factory.CreateConnection())
//                {
//                    connection.ClientId = "retriver-01";
//                    connection.Start();

//                    ISession session = connection.CreateSession();
//                    //通过会话创建一个消费者，这里就是Queue这种会话类型的监听参数设置
//                    var dest = session.GetDestination("first_queue", DestinationType.Queue);
//                    IMessageConsumer consumer = session.CreateConsumer(dest);
//                    //注册监听事件
//                    while (true)
//                    {
//                        var msg = consumer.Receive();

//                        var objMessage = msg as IObjectMessage;
//                        var user = objMessage.Body as UserInfo;
//                        Console.WriteLine(user);

//                    }
//                }
//            }

//            private static void consumer_Listener(IMessage message)
//            {
//                var objMessage = message as IObjectMessage;
//                var user = objMessage.Body as UserInfo;
//                Console.WriteLine(user);
//            }

//            [Serializable]
//            public class UserInfo
//            {
//                public int ID { get; set; }
//                public string Username { get; set; }

//                public override string ToString()
//                {
//                    return String.Format("[Id: {0}, Username: {1}", ID.ToString(), Username);
//                }
//            }

//            static void retriveMessage()
//            {
//                Uri uri = new Uri("activemq:tcp://localhost:61616");
//                NMSConnectionFactory factory = new NMSConnectionFactory("activemq:tcp://localhost:61616");
//                using (var conn = factory.CreateConnection())
//                {
//                    conn.Start();
//                    using (var session = conn.CreateSession(AcknowledgementMode.AutoAcknowledge))
//                    {
//                        var dest = session.GetDestination("first_queue", DestinationType.Queue);
//                        var consumer = session.CreateConsumer(dest);
//                        while (true)
//                        {
//                            var msg = consumer.Receive();

//                            if (msg != null && msg is ITextMessage)
//                            {
//                                var txtMsg = msg as ITextMessage;

//                                Console.WriteLine("收到消息：" + txtMsg.Text);

//                            } else
//                            {
//                                Console.WriteLine("未接收到消息！！");
//                            }

//                        }
//                    }
//                    conn.Close();
//                }
//            }
//        }
//    }

//}


//static void mutiple_thread_listen()
//        {
//            var listener = new TestListener();
//            listener.ClientId = "client";
//            listener.Init();

//            int threadNums = 2;
//            for (int i = 1; i <= threadNums; i++)
//            {
//                ThreadPool.QueueUserWorkItem(state =>
//                {
//                    listener.listen();
//                });
//            }
//        }

//        class TestListener
//        {
//            public string ClientId { get; set; }

//            public IConnection Connection { get; set; }

//            public void Init()
//            {
//                var factory = new ConnectionFactory("tcp://localhost:61616");
//                Connection = factory.CreateConnection();
//                Connection.Start();
//            }

//            public void Close()
//            {
//                Connection.Close();
//            }

//            public void listen()
//            {
//                Console.WriteLine("[thread-{0}] consumer begin listen....",
//                  Thread.CurrentThread.ManagedThreadId);

//                ISession session = Connection.CreateSession();
//                //通过会话创建一个消费者，这里就是Queue这种会话类型的监听参数设置
//                var dest = session.GetDestination("first_queue", DestinationType.Queue);
//                IMessageConsumer consumer = session.CreateConsumer(dest);
//                consumer.Listener += receiveMessage;
              


//                //Console.ReadLine();


//                //using (IConnection connection = factory.CreateConnection())
//                //{
//                //    connection.ClientId = ClientId;
//                //    connection.Start();

//                //    ISession session = connection.CreateSession();
//                //    //通过会话创建一个消费者，这里就是Queue这种会话类型的监听参数设置
//                //    var dest = session.GetDestination("first_queue", DestinationType.Queue);
//                //    IMessageConsumer consumer = session.CreateConsumer(dest);
//                //    consumer.Listener += receiveMessage;
//                //    Console.WriteLine("[thread-{0}] consumer begin listen....", 
//                //        Thread.CurrentThread.ManagedThreadId);

//                //    Console.ReadLine();

//                //}
//            }

//            private void receiveMessage(IMessage message)
//            {
//                var msg = message as ITextMessage;
//                Console.WriteLine("[thread-{0}] consumer from mq: {1}", Thread.CurrentThread.ManagedThreadId, msg.Text);
//            }
//        }
